package org.ethereum.net.dpos;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.ethereum.core.Blockchain;
import org.ethereum.facade.SyncStatus;
import org.ethereum.listener.CompositeEthereumListener;
import org.ethereum.net.dpos.DPOSMessage.ResponseCode;
import org.ethereum.net.dpos.ElectionNode.VoteInfo;
import org.ethereum.net.rlpx.Node;
import org.ethereum.net.server.Channel;
import org.ethereum.net.server.ChannelManager;
import org.ethereum.sync.SyncManager;
import org.ethereum.validator.BlockHeaderValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spongycastle.util.encoders.Hex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * 
 * DPoS protocol handler based on P2P protocol network.
 * 
 * Netty stack:
 * |------------ETH, BZZ, SHH, DPoS protocols---------|
 * |------------P2P protocol -------------------------|
 * |------------TCP & UDP ----------------------------|
 * 
 * @author shaolinwu 03/10/2018
 *
 * test gradle package function.
 * @author update by mark.ma 03/29/2018
 *
 */
@Component
@Sharable
public class DPOSHandler extends SimpleChannelInboundHandler<DPOSMessage> implements Runnable{

	private static final Logger logger = LoggerFactory.getLogger("dpos");
	
	public final static byte VERSION = 0;
	
	@Autowired
	Blockchain blockchain;
	
	@Autowired
	BlockGenerator generator;
	
	@Autowired
	BlockHeaderValidator headerValidator;
	
	@Autowired
	CompositeEthereumListener listener;
	
	@Autowired
	ChannelManager channelManager;
	
	ElectionNode electionNode;
	
	@Autowired
	SyncManager syncManager;
	
	private final ScheduledExecutorService reconnectScheduler;
	
	public DPOSHandler() {
		reconnectScheduler = Executors.newSingleThreadScheduledExecutor();
	}
	
	void setElectionNode(ElectionNode electionNode) {
		this.electionNode = electionNode;
	}
	
	/**
	 * this activate happens for every established channel.
	 */
	public void activate() {
        logger.info("DPOS protocol activated");
        
        run();
    }
	
	@Override
	public void run() {
		if (electionNode != null && !electionNode.isLeaderNode()) {
	    		RegisterCandidateRequest request = new RegisterCandidateRequest(electionNode.getCurrentNodeId(), true);
	    		Channel c = this.getElectionPeer(electionNode.getElectionLeader());
	    		if (c == null) {
	    			reconnectScheduler.schedule(this, 10, TimeUnit.SECONDS);
	    			logger.info("Election p2p channel is not established! unable to register myself as candicate, try 10 seconds later.");
	    			return;
	    		}
	    		c.getChannelPipeline().channel().writeAndFlush(request);
	    		logger.info("Registering myself as candicate...");
	    }
	}
	
	public void deactivate() {
		
	}
	
	/**
	 * follow P2P network handler.
	 */
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, DPOSMessage msg) throws Exception {
        switch (msg.getCommand()) {
        		case RegisterCandidateREQUEST:
        			// received the register from remote peer.
        			RegisterCandidateRequest registrRequest = (RegisterCandidateRequest)msg;
        			logger.info("ElectionServer received candidate info: {}}", Hex.toHexString(registrRequest.getCandidateId()));
        			electionNode.addCandidateNode(registrRequest);
        			ctx.channel().writeAndFlush(new RegisterCandidateResponse(registrRequest.getCandidateId(), true));
        			break;
        		case RegisterCandidateRESPONSE:
        			RegisterCandidateResponse registrResponse = (RegisterCandidateResponse)msg;
        			logger.info("ElectionClient confirmed as candidate.");
        			break;
        		case PACKAGE_REQUEST:
		    		PackageRequest request = (PackageRequest)msg;
		    		if (logger.isDebugEnabled())
		    			logger.debug("received package request: {}", request.toString());
		    		if (!Arrays.equals(electionNode.getCurrentNodeId(), request.getPresidentId())) {
		    			logger.warn("Packaging node Id does not match! Request: {}", request.toString());
		    			PackageResponse response = new PackageResponse(null, request.getRound(), 
		    					request.getPresidentId(), request.getElectionId(),
		    					DPOSMessage.ResponseCode.PACKAGE_VERIFY_FAILURE);
		    			sendPackageResponseToElectionNode(response);
		    			return;
		    		}
		    		// check the best block whether is synchronized or not.
				if (syncManager != null) {
					SyncStatus syncStatus = syncManager.getSyncStatus();
					if (syncStatus.getBlockBestKnown() > syncStatus.getBlockLastImported() + 1) {//allowed 1 blocks gap.
						logger.info("Failed to package block due to blocks syncing is not completed yet. {}", syncStatus.toString());
						PackageResponse response = new PackageResponse(null, request.getRound(), 
		    					request.getPresidentId(), request.getElectionId(),
		    					DPOSMessage.ResponseCode.PACKAGE_NOTSYNC);
		    				sendPackageResponseToElectionNode(response);
						return;
					}
				}
		    		sendPackageResponseToElectionNode(generator.generate(request));
		        break;
		    case PACKAGE_RESPONSE:
		    		PackageResponse	presponsed = ((PackageResponse)msg);
		    		if (logger.isDebugEnabled()) {
		    			logger.debug("received package response: {}", presponsed.toString());
		    		}
		    		if (!Arrays.equals(presponsed.getElectionId(), electionNode.getElectionLeader())) {
					logger.warn("Packaging election Id does not match! Elected president node performs bad, remove it from candicate list. Response:", presponsed.toString());
		    			return;
		    		}
		    		if (presponsed.getResponseCode() == ResponseCode.PACKAGE_EMPTY) {
		    			// it's empty package, reset voting pool. reset.
		    			logger.info("Packaging block is skipped due to there was no transaction found at the remote peer.");
		    			electionNode.currentVotingPool = null;
		    			return;
		    		}
		    		if (presponsed.getResponseCode() == ResponseCode.PACKAGE_NOTSYNC) {
		    			logger.info("Blocks syncing of Elected president has not completed yet. remove it from candicate list. Response: {}", presponsed.toString());
		    			electionNode.currentVotingPool.confirmSyncFailed(presponsed.getPresidentId());
		    			return;
		    		}
		    		if (presponsed.getResponseCode() != ResponseCode.SUCCESS) {
		    			logger.warn("Packaging response error! Elected president node performs bad, remove it from candicate list. Response: {}", presponsed.toString());
		    			return;
		    		}
		    		// got the new generated block and verify.
		    	    headerValidator.validateAndLog(presponsed.getBlockHeader(), logger);
		    	    List<Long> roundList = new ArrayList<Long>(1);
		    	    roundList.add(electionNode.currentVotingPool.round);
		    	    electionNode.currentVotingPool.confirmSync(roundList, presponsed.getPresidentId());
		        break;
            case VOTE_PRESIDENT_REQUEST:
            		VotePresidentRequest vrequest = (VotePresidentRequest)msg;
            		if (logger.isDebugEnabled())
            			logger.debug("received vote request: {}", vrequest.toString());
            		if (vrequest.getCandicateIds() == null || vrequest.getCandicateIds().length == 0) {
            			logger.warn("VotePresidentRequest is invalid: {}", vrequest.toString());
            			VotePresidentResponse vresponse = new VotePresidentResponse(vrequest.getRound(), (byte)0, 
            					vrequest.getElectionId(), ResponseCode.VOTE_VERIFY_FAILURE);
            			sendVoteResponseToElectionNode(vresponse);
                		return;
            		}
            		byte length = (byte)vrequest.getCandicateIds().length;
            		byte selectedPId = (byte)(Math.random()*length);//TODO: use this strategy by default.
            		if (logger.isDebugEnabled()) {
            			logger.debug("voted for president id: {}", selectedPId);
            		}
            		VotePresidentResponse vresponse = new VotePresidentResponse(vrequest.getRound(), selectedPId, vrequest.getElectionId(), ResponseCode.SUCCESS);
            		sendVoteResponseToElectionNode(vresponse);
                break;
            case VOTE_PRESIDENT_RESPONSE:
            		VotePresidentResponse vresponsed = (VotePresidentResponse)msg;
            		if (logger.isDebugEnabled())
            			logger.debug("received vote response: {}", vresponsed.toString());
        			if (vresponsed.getResponseCode() != ResponseCode.SUCCESS) {
        				logger.warn("VotePresidentResponse error with result: {}", vresponsed.toString());
            			return;
            		}
        			if (!Arrays.equals(vresponsed.getElectionId(), electionNode.getElectionLeader())) {
        				logger.warn("VotePresidentResponse election Id does not match with result: {}", vresponsed.toString());
            			return;
            		}
        			VoteInfo pool = electionNode.currentVotingPool;
        			if (pool != null) {
	        			if (pool.round != vresponsed.getRound()) {
	        				logger.warn("VotePresidentResponse round Id does not match current round! result: {}", vresponsed.toString());
	            			return;
	            		}
	        			pool.voteFor(vresponsed.getCandicate());
        			}
                break;
            case CONFIRMED_BLOCK_SYNC:
            		// we required the confirmation message from each candidate node for selecting the stable nodes in the next round.
            		ConfirmedSyncMessage confirmedMsg = (ConfirmedSyncMessage)msg;
            		if (electionNode.currentVotingPool != null) {
            			electionNode.currentVotingPool.confirmSync(confirmedMsg.getRounds(), confirmedMsg.getCandicateId());
            		}
            		break;
            case VOTE_PRESIDENT_BROADCAST:
            		if (logger.isDebugEnabled())
            			logger.debug("received vote brodacast: {}", ((VotedPresidentBroadcast)msg).toString());
            		electionNode.electedPresidents.put(((VotedPresidentBroadcast)msg).getRound(), (VotedPresidentBroadcast)msg);
                break;
            default:
                break;
        }
	}
	
	public boolean sendVoteRequest(VotePresidentRequest request) {
		boolean flag = false;
		for (Channel channel : channelManager.getActivePeers()) {
			channel.getChannelPipeline().channel().writeAndFlush(request);
			//logger.info("send vote ticket to {}", channel.toString());
			flag = true;
        }
		return flag;
	}
	
	private boolean sendPackageResponseToElectionNode(PackageResponse response) {
		Channel c = this.getElectionPeer(electionNode.getElectionLeader());
		if (c == null) {
			logger.warn("Election p2p channel does not exit! unable to send response to the election node.");
			return false;
		}
		c.getChannelPipeline().channel().writeAndFlush(response);
		return true;
	}
	
	private boolean sendVoteResponseToElectionNode(VotePresidentResponse response) {
		Channel c = this.getElectionPeer(electionNode.getElectionLeader());
		if (c == null) {
			logger.warn("Election p2p channel does not exit! unable to send response to the election node.");
			return false;
		}
		c.getChannelPipeline().channel().writeAndFlush(response);
		return true;
	}
	
	public boolean sendConfirmedSyncToElectionNode(ConfirmedSyncMessage response) {
		Channel c = this.getElectionPeer(electionNode.getElectionLeader());
		if (c == null) {
			logger.warn("Election p2p channel does not exit! unable to send response to the election node.");
			return false;
		}
		c.getChannelPipeline().channel().writeAndFlush(response);
		return true;
	}
	
	public boolean sendPackageRequest(PackageRequest response) {
		Channel c = getActivePeer(response.getPresidentNodeInfo());
		if (c == null) {
			logger.warn("Channel does not exit! unable to send packaging request to voted node {}.", response.getPresidentNodeInfo());
			return false;
		}
		c.getChannelPipeline().channel().writeAndFlush(response);
		return true;
	}
	
	public Channel getActivePeer(Node node) {
		if (node == null) {
			return null;
		}
		Channel activePeer = channelManager.getActivePeer(node.getId());
		if (activePeer == null || activePeer.isDisconnected()) {
			channelManager.connect(node);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			activePeer = channelManager.getActivePeer(node.getId());
		}
		return activePeer;
	}
	
	public Channel getElectionPeer(byte[] nodeId) {
		return getActivePeer(getActivePeerNode(nodeId));
	}
	
	public Node getActivePeerNode(byte[] nodeId) {
		return channelManager.getNodeById(nodeId);
	}
	
	public boolean isActivePeer(Node node) {
		Channel activePeer = channelManager.getActivePeer(node.getId());
		if (activePeer == null || activePeer.isDisconnected()) {
			return false;
		}
		return true;
	}
	
}
